package com.dayvisit;

import com.config.HadoopConfiguration;
import com.entity.Visit;
import com.service.VisitService;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.springframework.util.CollectionUtils;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;

public class HadoopDayVisitDrive {


    private static final String HDFS_URL = "hdfs://192.40.10.130:9000";

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
    }

    public static void hadoopDayVisit(String strHadoopInput, String strHadoopOutput, String strHadoopLocal, VisitService visitService) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", HDFS_URL);  // core-site.xml

        //1. 获取job
        Job job = Job.getInstance(conf);

        //2. 设置jar路劲
        job.setJarByClass(HadoopDayVisitDrive.class);

        //3. 关联mapper 和 reducer
        job.setMapperClass(DayVisitMapper.class);
        job.setReducerClass(DayVisitReducer.class);

        //4. 设置map 输出的kv
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5. 最终输出的kv
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6. 设置输入路劲和输出路径
        FileInputFormat.setInputPaths(job, new Path(strHadoopInput + HadoopConfiguration.getDayVisit()));
        FileOutputFormat.setOutputPath(job, new Path(strHadoopOutput + HadoopConfiguration.getDayVisit()));

        FileSystem fs = FileSystem.get(new URI(HDFS_URL), conf);
        //提前删除目录及文件
        fs.delete(new Path(strHadoopInput + HadoopConfiguration.getDayVisit()), true);
        //提前删除目录及文件
        fs.delete(new Path(strHadoopOutput + HadoopConfiguration.getDayVisit()), true);

        //提前上传文件
        fs.copyFromLocalFile(false, true,
                new Path(strHadoopLocal + "logs\\" + HadoopConfiguration.getDayVisit()),
                new Path(strHadoopInput + HadoopConfiguration.getDayVisit()));


        //7. 提交job
        boolean result = job.waitForCompletion(true);

        //将hdfs中的文件拷贝下来
        fs.copyToLocalFile(false,
                new Path(strHadoopOutput + HadoopConfiguration.getDayVisit() + "/part-r-00000"),
                new Path(strHadoopLocal + "result\\" + HadoopConfiguration.getDayVisit() + "\\result.txt"), true);

        //将数据写入visit表
        visitService.deleteVisit();
        List<String> strFileLines = FileUtils.readLines(new File(strHadoopLocal + "result\\" + HadoopConfiguration.getDayVisit() + "\\result.txt"), "utf-8");
        if (!CollectionUtils.isEmpty(strFileLines)) {
            for (String strLine : strFileLines) {
                String[] strArray = strLine.split("\t");
                Visit visit = new Visit(strArray[0], Integer.parseInt(strArray[1]));
                visitService.insertVisit(visit);
            }
        }
        fs.close();
    }
}
